package com.hkbigdata.processapi;

import com.hkbigdata.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author liuanbo
 * @creat 2024-05-14-19:08
 * @see 2194550857@qq.com
 */
public class Flink03_ProcessJoinFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator = env.socketTextStream("hadoop102", 9999)
                .flatMap(new FlatMapFunction<String, WaterSensor>() {
                    @Override
                    public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                        String[] split = value.split(",");
                        WaterSensor waterSensor = new WaterSensor(
                                split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                        out.collect(waterSensor);
                    }
                });

        SingleOutputStreamOperator<WaterSensor> waterSensorSingleOutputStreamOperator1 = env.socketTextStream("hadoop102", 8888)
                .flatMap(new FlatMapFunction<String, WaterSensor>() {
                    @Override
                    public void flatMap(String value, Collector<WaterSensor> out) throws Exception {
                        String[] split = value.split(",");
                        WaterSensor waterSensor = new WaterSensor(
                                split[0],
                                Long.valueOf(split[1]),
                                Integer.valueOf(split[2])
                        );
                        out.collect(waterSensor);
                    }
                });

        //1. 在processapi里面独有的定时器和侧输出流和状态编程
        //2.AbstractRichFunction 包含运行时上下文对象RuntimeContext() 和 open()生命周期方法
        //3.process api 定时器+侧输出流+状态编程+运行时上下文+生命周期方法

        //总结：1.子类包含父类的所有方法
        //2.父类是抽象类那么必须重写所有的抽象方法
        //3.如果说是接口的话，那么必须重新所有方法，无论是抽象还是非抽象方法
        //4.子类必须重写父类抽象方法，非抽象方法可以不用重写也可以重写。
        waterSensorSingleOutputStreamOperator.join(waterSensorSingleOutputStreamOperator1)
                .where(WaterSensor::getId)
                .equalTo(WaterSensor::getId)
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .apply(new JoinFunction<WaterSensor, WaterSensor, String>() {
                    @Override
                    public String join(WaterSensor first, WaterSensor second) throws Exception {
                        return "first" + first + "second" + second;
                    }
                }).print();


        env.execute();

    }
}
